package dm

import java.io.InputStream
import java.net.URL

import org.apache.spark.broadcast.Broadcast
import org.apache.spark.sql.{DataFrame, Dataset, Row, SparkSession}
import org.apache.spark.{SparkConf, SparkContext}

import scala.io.{BufferedSource, Source}


object dm_bi_student_study_condition {
  def main(args: Array[String]): Unit = {

    val conf: SparkConf = new SparkConf().setAppName("dm.dm_bi_student_study_condition")
      .set("spark.sql.autoBroadcastJoinThreshold","1073741824")


    val spark: SparkSession = SparkSession.builder().config(conf).getOrCreate()

    import spark.implicits._

    //1. 清除dm_bi_student_study_condition表数据
    spark.sql("truncate table dm.dm_bi_student_study_condition")

    //2. 执行插入语句
    val result_date: DataFrame = spark.sql(
      """
          SELECT
                  `a`.`college_id`,
                  `a`.`service_end_date`,
                  from_unixtime(unix_timestamp(min(to_date(`a`.`jiesuanri`))) + 3600 * 12) `result_date`
           FROM
             (SELECT `a`.`college_id`,
                     `a`.`service_end_date`,
                     `c`.`date_type`,
                     CASE
                         WHEN `a`.`college_id` = 'j5m484vz' THEN lead(`c`.`date_type`, 9) OVER (PARTITION BY `a`.`college_id`, `a`.`service_end_date`
                                                                                        ORDER BY `c`.`date_type` ASC)
                         WHEN `a`.`college_id` = 'j5m48deg' THEN lead(`c`.`date_type`, 10) OVER (PARTITION BY `a`.`college_id`, `a`.`service_end_date`
                                                                                         ORDER BY `c`.`date_type` ASC)
                         WHEN `a`.`college_id` = 'j5m484sh' THEN lead(`c`.`date_type`, 12) OVER (PARTITION BY `a`.`college_id`, `a`.`service_end_date`
                                                                                         ORDER BY `c`.`date_type` ASC)
                         WHEN `a`.`college_id` = 'jukl2lsoll' THEN lead(`c`.`date_type`, 12) OVER (PARTITION BY `a`.`college_id`, `a`.`service_end_date`
                                                                                         ORDER BY `c`.`date_type` ASC)
                         WHEN `a`.`college_id` = 'k4xgnlldsp' THEN lead(`c`.`date_type`, 10) OVER (PARTITION BY `a`.`college_id`, `a`.`service_end_date`
                                                                                         ORDER BY `c`.`date_type` ASC)
                     END `jiesuanri`
              FROM
                (SELECT DISTINCT `dim_xf_t_class`.`college_id`,
                                 `dim_xf_t_class`.`service_end_date`
                 FROM `dim`.`dim_xf_t_class`) `a`
              LEFT OUTER JOIN
                (SELECT `dim_date`.`date_type`
                 FROM `dim`.`dim_date`
                 WHERE `dim_date`.`statutory_holiday_flag` = '0') `c` ON 1 = 1
              WHERE to_date(`a`.`service_end_date`) <= to_date(`c`.`date_type`)
                AND to_date(`c`.`date_type`) < date_add(to_date(`a`.`service_end_date`), 30)) `a`
           GROUP BY `a`.`college_id`,`a`.`service_end_date`
        """)
    result_date.createOrReplaceTempView("result_date")
    result_date.cache()

    spark.sql(
      """
        insert into table  dm.dm_bi_student_study_condition partition(dt= '0000-00-00')
        SELECT `a`.`class_id`,
                           `a`.`class_name`,
                           `a`.`student_id`,
                           `c`.`stage_id`,
                           `d`.`result_date`,
                           `b`.`college_id`,
                           `b`.`course_title`,
                           `b`.`channel_id`,
                           `b`.`channel_title`,
                           `b`.`teacher_id`,
                           `b`.`listen_m`,
                           `b`.`audition_date`,
                           `b`.`course_nature_type`,
                           `b`.`curriculum_id`,
                           `e`.`jobcompletion_id`
                    ,'Y' flag
                    ,from_unixtime(unix_timestamp(),'yyyy-MM-dd HH:mm:ss') op_time
                    ,'$exe_id' execid
                    FROM `dws`.`dws_xf_t_class_change_record` `a`
                    LEFT OUTER JOIN `dws`.`dws_attend_lecturesn` `b` ON `a`.`student_id` = `b`.`dapeng_user_id`
                    AND `a`.`college_id` = `b`.`college_id`
                    LEFT OUTER JOIN `dim`.`dim_xf_t_class` `c` ON `a`.`class_id` = `c`.`class_id`
                    LEFT OUTER JOIN result_date d ON `c`.`college_id` = `d`.`college_id`
                    AND `c`.`service_end_date` = `d`.`service_end_date`
                    LEFT OUTER JOIN `dws`.`dws_rev_jobcompletion` `e` ON `a`.`student_id` = `e`.`user_id`
                    AND `b`.`curriculum_id` = `e`.`curriculum_id`
                    WHERE NOT `b`.`curriculum_title` LIKE '%作业点评%'
                      AND NOT `b`.`curriculum_title` LIKE '%开班大会%'
                      AND NOT `b`.`curriculum_title` LIKE '%课前预热%'
                      AND NOT `b`.`curriculum_title` LIKE '43'
                      and `b`.`curriculum_type` = 'OFFICIAL'
                      AND (`b`.`first_listen_time` BETWEEN `c`.`service_start_date` AND `d`.`result_date`)
                      AND `a`.`learn_status` NOT IN ('NOT_ENROLLED', 'ENROLLED', 'DROPPED_SCHOOL')
        """)

    spark.stop()

  }
}
